package com.mxd.flink.connector.table;

import com.mxd.flink.connector.config.RedisOptions;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.*;

import java.util.HashSet;
import java.util.Set;

import static org.apache.flink.table.factories.FactoryUtil.createTableFactoryHelper;

/**
 * 实现sink和source工厂接口的工厂类
 * @author rongdi
 * @date 2022/9/18 13:21
 */
public class RedisDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {


    /**
     * 创建DynamicTableSink对象
     * @param context
     * @return
     */
    @Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        /**
         * 拿到一个帮助类，主要作用是为了拿到flinksql中建表语句with里参数的帮助类
         */
        FactoryUtil.TableFactoryHelper helper = createTableFactoryHelper(this, context);

        /**
         * 拿到with里的所有参数
         */
        ReadableConfig config = helper.getOptions();
        /**
         * 校验参数
         */
        validateOptions(config);
        EncodingFormat<SerializationSchema<RowData>> encodingFormat = null;
        /**
         * 这里判断format有没有是因为，下面helper.discoverDecodingFormat会要求format参数必填，
         * 如果不判断，当redis作为维表的时候很可能不需要传format，这里就报错了
         */
        if(config.get(RedisOptions.FORMAT) != null) {
            /**
             * 根据connector的format配置项，拿到EncodingFormat对象，如json格式需要依赖flink-json包
             */
            encodingFormat = helper.discoverEncodingFormat(SerializationFormatFactory.class,FactoryUtil.FORMAT);
        }
        helper.validate();
        return new RedisDynamicTableSink(context.getCatalogTable().getResolvedSchema(), config, encodingFormat);
    }

    /**
     * 创建DynamicTableSource对象
     * @param context
     * @return
     */
    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        /**
         * 拿到一个帮助类，主要作用是为了拿到flinksql中建表语句with里参数的帮助类
         */
        FactoryUtil.TableFactoryHelper helper = createTableFactoryHelper(this, context);

        /**
         * 拿到with里的所有参数
         */
        ReadableConfig options = helper.getOptions();
        /**
         * 校验参数
         */
        validateOptions(options);
        DecodingFormat<DeserializationSchema<RowData>> decodingFormat = null;
        /**
         * 这里判断format有没有是因为，下面helper.discoverDecodingFormat会要求format参数必填，
         * 如果不判断，当redis作为维表的时候很可能不需要传format，这里就报错了
         */
        if(options.get(RedisOptions.FORMAT) != null) {
            /**
             * 根据connector的format配置项，拿到DecodingFormat对象，如json格式需要依赖flink-json包
             */
            decodingFormat = helper.discoverDecodingFormat(DeserializationFormatFactory.class, FactoryUtil.FORMAT);
        }
        /**
         * 校验配置
         */
        helper.validate();
        return new RedisDynamicTableSource(options, context.getCatalogTable().getResolvedSchema(), decodingFormat);
    }

    /**
     * 指定工厂类的标识符,也就是建表时必须填写的connector参数的值
     * @return
     */
    @Override
    public String factoryIdentifier() {
        return "redis";
    }

    /**
     * 使用该connector必填的参数集合
     * @return
     */
    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> options = new HashSet();
        options.add(RedisOptions.HOST);
        options.add(RedisOptions.PORT);
        options.add(RedisOptions.REDISMODE);
        return options;
    }

    /**
     * 使用该connector可选的参数集合
     * @return
     */
    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> options = new HashSet();
        options.add(RedisOptions.DATABASE);
        options.add(RedisOptions.HOST);
        options.add(RedisOptions.PORT);
        options.add(RedisOptions.MAXIDLE);
        options.add(RedisOptions.MAXTOTAL);
        options.add(RedisOptions.CLUSTERNODES);
        options.add(RedisOptions.PASSWORD);
        options.add(RedisOptions.TIMEOUT);
        options.add(RedisOptions.MINIDLE);
        options.add(RedisOptions.COMMAND);
        options.add(RedisOptions.KEY);
        options.add(RedisOptions.FORMAT);
        options.add(RedisOptions.BATCH_FETCH_ROWS);
        options.add(RedisOptions.REDISMODE);
        options.add(RedisOptions.LOOKUP_CACHE_MAX_ROWS);
        options.add(RedisOptions.LOOKUP_CHCHE_TTL);
        options.add(RedisOptions.LOOKUP_MAX_RETRIES);
        options.add(RedisOptions.LOOKUP_CACHE_LOAD_ALL);
        options.add(RedisOptions.LOOKUP_ASYNC_ENABLED);
        options.add(RedisOptions.SINK_MAX_RETRIES);
        options.add(RedisOptions.SINK_PARALLELISM);
        options.add(RedisOptions.SINK_KEY_TTL);
        return options;
    }

    /**
     * 校验各个参数
     * @param options
     */
    private void validateOptions(ReadableConfig options) {
    }
}
